package org.robby.hbase.cdr;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

public class HbaseCdrIF {
	HTablePool pool;
	public static HbaseCdrIF hbsase = null;

	private static String tab_cdr_name = "tab_cdr";
	private static String tab_cdr_daily_name = "tab_cdr_daily";

	private HbaseCdrIF() {
		HBaseConfiguration config = new HBaseConfiguration();
		config.set("hbase.zookeeper.quorum", "192.168.1.90");// 连接到远程服务上
		pool = new HTablePool(config, 1000);
	}

	public static HbaseCdrIF getInstance() {
		if (hbsase == null) {
			hbsase = new HbaseCdrIF();
		}
		return hbsase;
	}

	public List<CdrPro.SmCdr> getSmCdr(String addr) throws Exception {
		List<CdrPro.SmCdr> l = new ArrayList<CdrPro.SmCdr>();
		HTableInterface tab_cdr = pool.getTable(tab_cdr_name);
		Scan s = new Scan();
		s.setStartRow(addr.getBytes());// 大于等于START值
		byte[] stop = addr.getBytes();// 对start加1作为截止查询数据。
		stop[stop.length - 1] += 1;
		s.setStopRow(stop);// 小于STOP值
		ResultScanner rs = tab_cdr.getScanner(s);
		for (Result r : rs) {
			byte[] data = r.getValue(Bytes.toBytes("data"), null);
			CdrPro.SmCdr sm = CdrPro.SmCdr.parseFrom(data);
			l.add(sm);

		}
		return l;
	}

	public Map<String, Integer> getDailyReport() throws Exception {
		Map<String, Integer> m = new LinkedHashMap<String, Integer>();
		HTableInterface tab_cdr_daily = pool.getTable(tab_cdr_daily_name);
		Scan scan = new Scan();
		ResultScanner rs = tab_cdr_daily.getScanner(scan);
		for (Result r : rs) {
			byte[] key = r.getRow();
			byte[] value = r.getValue(Bytes.toBytes("data"), null);
			m.put(Bytes.toString(key), Bytes.toInt(value));
		}
		return m;
	}

	public static void main(String[] args) throws Exception {
		HbaseCdrIF hif = HbaseCdrIF.getInstance();
		List<CdrPro.SmCdr> l = hif.getSmCdr("13900001711");
		for (CdrPro.SmCdr sm : l) {
			System.out.println(sm.toString());
		}
	}
}
